﻿using CsvHelper;
using FastEmit;
using Flurl.Http;
using OkFlurl;
using System.Globalization;
using System.IO.Compression;
using System.Net.Http;
using System.Runtime.CompilerServices;
using System.Text.Json;
using System.Text.Json.Nodes;

namespace InfluxDB.Flurl
{
    //https://docs.influxdata.com/influxdb/v2/reference/syntax/line-protocol/#naming-restrictions
    //https://docs.influxdata.com/influxdb/v2/api/#operation/PostWrite
    //https://docs.influxdata.com/influxdb/v2/write-data/developer-tools/api/
    //https://docs.influxdata.com/influxdb/v2/write-data/best-practices/optimize-writes/#use-gzip-compression
    //https://docs.influxdata.com/influxdb/v2/reference/syntax/line-protocol/#naming-restrictions
    //https://docs.influxdata.com/flux/v0/get-started/
    public class InfluxDbClient
    {
        #region static

        private static string GetPrecision(WritePrecision precision)
        {
            var val = precision switch
            {
                WritePrecision.Ms => "ms",
                WritePrecision.S => "s",
                WritePrecision.Us => "us",
                WritePrecision.Ns => "ns",
                _ => "ns"
            };
            return val;
        }



        #endregion

        #region primary

        private readonly string _token;

        public FlurlApi Api { get; }

        public FlurlRequest CreateRequest(string path)
        {
            var req = Api.CreateRequest(path);
            req.WithHeader("Authorization", _token);
            return req;
        }

        #endregion

        public InfluxDbClient(string url, string token)
        {
            Api = new(url);
            _token = $"Token {token}";
        }

        public async Task CreateOrgAsync(string name, string description = null, CancellationToken token = default)
        {
            var req = CreateRequest("/api/v2/orgs");
            await Api.PostAsync(req, new { name, description }, token: token);
        }

        public async Task<List<OrgEntity>> GetOrgListAsync(int limit, int offset = 0, string org = null, string orgId = null, string userID = null, CancellationToken token = default)
        {
            if (limit < 1 || limit > 100)
            {
                throw new Exception("limit must be between 1 and 100");
            }

            if (offset < 0)
            {
                throw new Exception("offset must >=0");
            }

            var req = CreateRequest("/api/v2/orgs");
            req.AppendQueryParam("limit", limit);
            req.AppendQueryParam("offset", offset);

            if (!string.IsNullOrEmpty(org))
            {
                req.AppendQueryParam("org", org);
            }
            if (!string.IsNullOrEmpty(orgId))
            {
                req.AppendQueryParam("orgId", orgId);
            }
            if (!string.IsNullOrEmpty(userID))
            {
                req.AppendQueryParam("userID", userID);
            }

            var res = await Api.GetJsonAsync<JsonObject>(req, null, token);
            return res["orgs"].Deserialize<List<OrgEntity>>();
        }

        public async Task<List<OrgEntity>> GetOrgListAsync(CancellationToken token = default)
        {
            var limit = 100;
            var offset = 0;
            var list = new List<OrgEntity>();
            while (true)
            {
                var res = await GetOrgListAsync(limit, offset, token: token);
                if (res == null || res.Count <= 0)
                {
                    break;
                }
                list.AddRange(res);
                offset += limit;
            }
            return list;
        }

        public async Task<bool> ExistsOrgAsync(string name, CancellationToken token = default)
        {
            var list = await GetOrgListAsync(token);
            return list.Any(a => a.name == name);
        }

        public async Task<OrgEntity> GetOrgAsync(string name, CancellationToken token = default)
        {
            var res = await GetOrgListAsync(1, 0, name, null, null, token);
            if (res == null || res.Count <= 0)
            {
                return null;
            }
            return res[0];
        }

        public async Task DeleteOrgByIdAsync(string id, CancellationToken token = default)
        {
            var req = CreateRequest($"/api/v2/orgs/{id}");
            await req.DeleteAsync(cancellationToken: token);
        }

        public async Task DeleteOrgAsync(string orgName, CancellationToken token = default)
        {
            var org = await GetOrgAsync(orgName, token);
            if (org == null)
            {
                throw new Exception($"{orgName} not exists.");
            }
            await DeleteOrgByIdAsync(org.id, token);
        }

        public async Task CreateBucketByOrgIdAsync(string bucketName, string orgId, int expireHour, CancellationToken token = default)
        {
            var req = CreateRequest("/api/v2/buckets");
            var entity = new BucketEntity
            {
                name = bucketName,
                orgID = orgId,
                retentionRules =
                [
                    new()
                    {
                        everySeconds = expireHour*3600,
                        shardGroupDurationSeconds = expireHour*3600
                    }
                ]
            };
            await Api.PostAsync(req, entity, token: token);
        }

        public async Task CreateBucketAsync(string bucketName, string orgName, int expireHour, CancellationToken token = default)
        {
            var orgList = await GetOrgListAsync(1, org: orgName, token: token);
            if (orgList == null || orgList.Count <= 0)
            {
                throw new Exception($"{orgName} not exist.");
            }
            await CreateBucketByOrgIdAsync(bucketName, orgList[0].id, expireHour, token);
        }

        public async Task<List<BucketEntity>> GetBucketListAsync(string org, int limit, int offset = 0, string bucketName = null, CancellationToken token = default)
        {
            if (limit < 1 || limit > 100)
            {
                throw new Exception("limit must be between 1 and 100");
            }

            if (offset < 0)
            {
                throw new Exception("offset must >=0");
            }

            var req = CreateRequest("/api/v2/buckets");
            req.AppendQueryParam("org", org);
            req.AppendQueryParam("limit", limit);
            req.AppendQueryParam("offset", offset);
            if (!string.IsNullOrEmpty(bucketName))
            {
                req.AppendQueryParam("name", bucketName);
            }
            var res = await Api.GetJsonAsync<JsonObject>(req, null, token);
            return res["buckets"].Deserialize<List<BucketEntity>>();
        }

        public async Task<List<BucketEntity>> GetBucketListAsync(string org, CancellationToken token = default)
        {
            var limit = 100;
            var offset = 0;
            var list = new List<BucketEntity>();
            while (true)
            {
                var res = await GetBucketListAsync(org, limit, offset, token: token);
                if (res == null || res.Count <= 0)
                {
                    break;
                }
                list.AddRange(res);
                offset += limit;
            }
            return list;
        }

        public async Task<bool> ExistsBucketAsync(string bucketName, string orgName, CancellationToken token = default)
        {
            var list = await GetBucketListAsync(orgName, token);
            return list.Any(a => a.name == bucketName);
        }

        public async Task<BucketEntity> GetBucketAsync(string bucketName, string orgName, CancellationToken token = default)
        {
            var list = await GetBucketListAsync(orgName, 1, 0, bucketName, token);
            if (list == null || list.Count <= 0)
            {
                return null;
            }
            return list[0];
        }

        public async Task DeleteBucketByIdAsync(string id, CancellationToken token = default)
        {
            var req = CreateRequest($"/api/v2/buckets/{id}");
            await req.DeleteAsync(cancellationToken: token);
        }

        public async Task DeleteBucketAsync(string bucketName, string orgName, CancellationToken token = default)
        {
            var bucket = await GetBucketAsync(bucketName, orgName, token);
            if (bucket == null)
            {
                throw new Exception($"{bucketName} not exist.");
            }
            await DeleteBucketByIdAsync(bucket.id, token);
        }

        public async Task WriteAsync(Func<Stream, Task> func, string bucket, string org, WritePrecision? precision = null, bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, CancellationToken token = default)
        {
            var req = CreateRequest("/api/v2/write");
            req.AppendQueryParam("bucket", bucket);
            req.AppendQueryParam("org", org);
            if (precision.HasValue)
            {
                req.AppendQueryParam("precision", GetPrecision(precision.Value));
            }

            req.WithHeader("Accept", "application/json");

            if (!gzip)
            {
                req.WithHeader("Content-Type", "text/plain; charset=utf-8");

            }
            else
            {
                req.WithHeader("Content-Type", "application/x-gzip");
                req.WithHeader("Content-Encoding", "gzip");
                req.WithHeader("Accept-Encoding", "identity");
            }

            using var fcs = new FlurlStreamContent(async (stream) =>
            {
                if (!gzip)
                {
                    await func(stream);
                    return;
                }
                using var gz = new GZipStream(stream, compressLv, true);
                await func(gz);
            });

            (await Api.PostAsync(req, fcs, token: token)).Dispose();
        }

        public Task WriteAsync<T>(IEnumerable<T> list, string measurement, string bucket, string org, WritePrecision? precision = null, bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, int sendBuffer = 100, CancellationToken token = default) where T : class
        {
            var t = typeof(T);
            if (t == typeof(object))
            {
                t = list.First().GetType();
            }
            var accssor = FastEmitFactory.GetTypeAccessor(t);
            var ch = CacheHelper.GetCache(t);

            return WriteAsync(async stream =>
            {
                var writer = new InfluxDbWriter(stream, token);
                var i = 0;
                var currentRead = 0;
                foreach (var item in list)
                {
                    i++;
                    currentRead++;

                    if (i > 1)
                    {
                        await writer.WriteNAsync();
                    }

                    await writer.WriteAsync(measurement, item, accssor, ch);

                    if (currentRead == sendBuffer)
                    {
                        currentRead = 0;
                        await stream.FlushAsync(token);
                    }
                }

            }, bucket, org, precision, gzip, compressLv, token);
        }

        public Task WriteLinesAsync(IEnumerable<string> list, string bucket, string org, WritePrecision? precision = null, bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, int sendBuffer = 100, CancellationToken token = default)
        {
            return WriteAsync(async stream =>
            {
                var writer = new InfluxDbWriter(stream, token);
                var i = 0;
                var currentRead = 0;
                foreach (var item in list)
                {
                    i++;
                    currentRead++;
                    if (i > 1)
                    {
                        await writer.WriteNAsync();
                    }

                    await writer.WriteLineAsync(item);

                    if (currentRead == sendBuffer)
                    {
                        currentRead = 0;
                        await stream.FlushAsync(token);
                    }
                }

            }, bucket, org, precision, gzip, compressLv, token);
        }

        public Task WritePointDataAsync(IEnumerable<PointData> list, string bucket, string org, WritePrecision? precision = null, bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, int sendBuffer = 100, CancellationToken token = default)
        {
            return WriteAsync(async stream =>
            {
                var writer = new InfluxDbWriter(stream, token);
                var i = 0;
                var currentRead = 0;
                foreach (var item in list)
                {
                    i++;
                    currentRead++;
                    if (i > 1)
                    {
                        await writer.WriteNAsync();
                    }

                    await writer.WritePointDataAsync(item);

                    if (currentRead == sendBuffer)
                    {
                        currentRead = 0;
                        await stream.FlushAsync(token);
                    }
                }

            }, bucket, org, precision, gzip, compressLv, token);
        }

        public async Task<Stream> QueryStreamAsync(string query, string org, bool gzip = false, CancellationToken token = default)
        {
            var req = CreateRequest("/api/v2/query");
            req.AppendQueryParam("org", org);
            req.WithHeader("Content-Type", "application/vnd.flux");
            req.WithHeader("Accept", "application/csv");
            if (gzip)
            {
                req.WithHeader("Accept-Encoding", "gzip");
            }

            var res = await req.PostStringAsync(query, HttpCompletionOption.ResponseHeadersRead, token);
            var stream = await res.GetStreamAsync();

            var isgz = res.Headers.Any(a => a.Name == "Content-Encoding" && "gzip".Equals(a.Value, StringComparison.OrdinalIgnoreCase));

            if (!isgz)
            {
                return stream;
            }

            return new GZipStream(stream, CompressionMode.Decompress);
        }

        public async IAsyncEnumerable<InfluxDbRecord> QueryUnbufferAsync(string query, string org, bool gzip = false, [EnumeratorCancellation] CancellationToken token = default)
        {
            using var stream = await QueryStreamAsync(query, org, gzip, token);
            using var sr = new StreamReader(stream);
            using var csv = new CsvReader(sr, CultureInfo.InvariantCulture);

            List<string> cols = null;
            var i = 0;
            var hasTable = false;
            var tableIndex = 0;
            while (await csv.ReadAsync())
            {
                token.ThrowIfCancellationRequested();

                if ("error".Equals(csv[1]) && "reference".Equals(csv[2]))
                {
                    throw new Exception(csv.Parser.RawRecord);
                }

                i++;

                if (i == 1) //第一行
                {
                    cols = csv.Parser.Record.ToList();
                    if (csv.Parser.Record.Length > 2)
                    {
                        if (csv[2] == "table")
                        {
                            hasTable = true;
                        }
                    }
                }
                else
                {
                    if (hasTable)
                    {
                        var ok = int.TryParse(csv[2], out var index);
                        if (!ok)
                        {
                            continue;
                        }
                        tableIndex = index;
                    }

                    var record = new InfluxDbRecord
                    {
                        Table = tableIndex,
                        Keys = cols
                    };

                    for (var j = 0; j < record.Keys.Count; j++)
                    {
                        record.Values.Add(record.Keys[j], csv[j]);
                    }

                    yield return record;
                }
            }
        }

        public async Task<List<InfluxDbRecord>> QueryAsync(string query, string org, bool gzip = false, CancellationToken token = default)
        {
            List<InfluxDbRecord> list = [];
            var unBuffer = QueryUnbufferAsync(query, org, gzip, token);
            await foreach (var record in unBuffer)
            {
                list.Add(record);
            }
            return list;
        }

        public async Task<List<T>> QueryAsync<T>(string query, string org, bool gzip = false, CancellationToken token = default) where T : class, new()
        {
            Dictionary<DateTimeOffset, T> _dict = [];
            var t = typeof(T);
            var ch = CacheHelper.GetCache(t);
            var accssor = new TypeAccessor(t);
            var unBuffer = QueryUnbufferAsync(query, org, gzip, token);
            await foreach (var record in unBuffer)
            {
                var time = record.GetTime();
                var ok = _dict.TryGetValue(time, out var model);
                if (!ok)
                {
                    model = new();

                    //set time
                    accssor.SetValue(model, ch.TimeKey, time.ToLocalTime());

                    //set tags
                    foreach (var item in ch.TagKeys)
                    {
                        accssor.SetValueConvert(model, item, record.GetValueByKey(item));
                    }
                    _dict.Add(time, model);
                }

                //set values
                var key = record.GetField();
                if (accssor.MemberDict.ContainsKey(key))
                {
                    var val = record.GetValue();
                    accssor.SetValueConvert(model, key, val);
                }
            }
            return _dict.Select(s => s.Value).ToList();
        }
    }
}
